package com.atguigu.edu.realtime.app.dwd.db;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.userLogin;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

public class DwdUserLogin {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        //从kafka topic_log里面读取数据
        String topic = "dwd_traffic_page_log";
        String groupId = "dwd_user_login";
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        DataStreamSource<String> kafkaDs = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");

        //过滤页面日志，uid日志不为空
        SingleOutputStreamOperator<JSONObject> pageDs = kafkaDs.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                if (jsonObject.getJSONObject("common").getString("uid") != null) {
                    out.collect(jsonObject);
                }
            }
        });

        //指定水位线
        SingleOutputStreamOperator<JSONObject> wmDs = pageDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner((a, b) -> Long.parseLong(a.getString("ts")))
        );
        //按照会话分组
        KeyedStream<JSONObject, String> keyedDS = wmDs.keyBy(a -> a.getJSONObject("common").getString("mid"));
        //用状态，取第一个数据
        SingleOutputStreamOperator<JSONObject> processDs = keyedDS.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            private ValueState<JSONObject> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<JSONObject> stateDescriptor = new ValueStateDescriptor<JSONObject>("valueState", JSONObject.class);
                stateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1))
                        //.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        .build());
                valueState = getRuntimeContext().getState(stateDescriptor);
            }

            @Override
            public void processElement(JSONObject value, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                JSONObject valueSta = valueState.value();
                if (valueSta == null) {
                    valueState.update(value);
                    long time = ctx.timerService().currentProcessingTime();
                    ctx.timerService().registerProcessingTimeTimer(time + 10000);
                } else {
                    String ts = valueSta.getString("ts");
                    String tsvs = value.getString("ts");
                    if (tsvs.compareTo(ts) < 0) {
                        valueState.update(value);
                    }
                }
            }

            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                out.collect(valueState.value());
            }
        });
        //转换数据  为了方便的赋值，但是写入kafka的时候还得是String
        SingleOutputStreamOperator<String> userDs = processDs.map(new MapFunction<JSONObject, String>() {
            @Override
            public String map(JSONObject value) throws Exception {
                JSONObject common = value.getJSONObject("common");
                Long ts = value.getLong("ts");
                String loginTime = DateFormatUtil.toYmdHms(ts);
                String logindate = loginTime.substring(0, 10);
                userLogin userLog = userLogin.builder()
                        .userId(common.getString("uid"))
                        .loginTime(loginTime)
                        .date(logindate)
                        .ar(common.getString("ar"))
                        .ch(common.getString("ch"))
                        .mid(common.getString("mid"))
                        .ts(ts)
                        .build();
                return JSON.toJSONString(userLog);
            }
        });
        userDs.print();
        //输出数据
        userDs.sinkTo(MyKafkaUtil.getKafkaSink("dwd_user_login"));

        env.execute();
    }
}


